package top.v5it.logback.push.convergence;

import cn.hutool.core.util.NumberUtil;
import lombok.Getter;
import lombok.Setter;

import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 固定时间计数收敛策略，在固定时间内累计发生次数达到计数限制发送
 *
 * @author zhanpu
 * @date 2022/9/9
 */
@Setter
@Getter
public class FixeTimeCountConvergenceStrategy implements ConvergenceStrategy {

    private final static ConcurrentMap<String, AtomicLong> counter = new ConcurrentHashMap<>();
    private final static ConcurrentMap<String, Long> delay = new ConcurrentHashMap<>();

    /**
     * 限制大小
     */
    private long limit;

    /**
     * 超时时间，单位秒
     */
    private long timeout;

    /**
     * 在固定单位时间内按消息体中消息编号统计单条消息到达数量，当单条消息数量达到{@link FixeTimeCountConvergenceStrategy#limit}限制就推送此条消息并删除计数器和超时间，
     * 当单条消息数量未达到{@link FixeTimeCountConvergenceStrategy#limit}限制并且超过{@link FixeTimeCountConvergenceStrategy#timeout}限制，按策略丢弃部分消息等待下个统计周期
     *
     * @param payload 消息体
     * @return
     */
    @Override
    public boolean isPush(byte[] payload) {
        return getMsgId(payload).map(t -> {
            final long toEpochSecond = ConvergenceStrategy.toEpochSecond();
            // 首次不推送
            if (delay.putIfAbsent(t, toEpochSecond) == null &&
                    counter.putIfAbsent(t, new AtomicLong(1)) == null) return false;
            final AtomicLong value = counter.computeIfPresent(t, (k, v) -> {
                v.incrementAndGet();
                final Long time = delay.get(k);
                // 达到限制
                if (v.compareAndSet(limit, 0)) return null;
                // 超过时间未达到，减去部分统计数值
                if ((time + timeout - toEpochSecond) <= 0) {
                    final long diff = toEpochSecond - time - timeout;
                    v.accumulateAndGet(diff, (l, r) -> {
                        final double rate = limit != timeout ? NumberUtil.div(limit, timeout, 2) : 0.30D;
                        final long quantity = (long) NumberUtil.mul(r, rate);
                        return l - quantity;
                    });
                }
                return v;
            });
            if (Objects.nonNull(value)) {
                delay.replace(t, toEpochSecond);
                return false;
            }
            delay.remove(t);
            return true;
        }).orElse(Boolean.FALSE);
    }
}
